package lab82;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * AllCityMapper和SomeCityMapper经过Shuffle之后的输出格式如下： <济南，{a_济南，s_济南 济南飞机场}>
 * <德州，{a_德州}> <青岛，{a_青岛，s_青岛 青岛飞机场}> ....
 * 如果集合中有飞机场，那这个集合中的所有城市将会被过滤，把集合中没有飞机场的城市进行输出就可以了
 */

public class CityReduce extends Reducer<Text, Text, Text, Text> {

    public void reduce(Text key, java.lang.Iterable<Text> values,
            org.apache.hadoop.mapreduce.Reducer<Text, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {

        // 定义城市的名字，将其值初始化为null
        String cityName = null;
        // 定义存城市的集合list
        List<String> list = new ArrayList<String>();
        // 开始循环遍历集合中的元素
        for (Text value : values) {
            // 判断集合中的元素如果是以s_开头的，则对cityName进行初始化
            // 否则将该城市名添加到list集合中
            if (value.toString().startsWith(SomeCityMapper.LABEL)) {
                int index = value.toString().indexOf("_");
                cityName = value.toString().substring(index + 1, index + 3);
            } else if (value.toString().startsWith(AllCityMapper.LABEL)) {
                list.add(value.toString().substring(2));
            }
        }
        /**
         * s_开头的都是有飞机场的城市，所有若cityName为null且list.size大于0，则是
         * 没有飞机场的城市，就是我们要过滤的城市，将其进行输出就可以了
         */
        if (cityName == null && list.size() > 0) {
            for (String str : list) {
                context.write(new Text(str), new Text(""));
            }
        }
    }
}